package com.bigfans.searchservice.listener;

import com.bigfans.framework.es.ElasticTemplate;
import com.bigfans.framework.es.IndexDocument;
import com.bigfans.framework.es.request.CreateDocumentCriteria;
import com.bigfans.model.event.CategoryCreatedEvent;
import com.bigfans.framework.kafka.KafkaConsumerBean;
import com.bigfans.framework.kafka.KafkaListener;
import com.bigfans.searchservice.api.clients.CatalogServiceClient;
import com.bigfans.searchservice.model.Category;
import com.bigfans.searchservice.schema.mapping.CategoryMapping;
import com.bigfans.searchservice.schema.mapping.ProductMapping;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * @author lichong
 * @create 2018-02-10 下午2:03
 **/
@KafkaConsumerBean
@Component
public class CategoryListener {

    @Autowired
    private CatalogServiceClient catalogServiceClient;
    @Autowired
    private ElasticTemplate elasticTemplate;

    @KafkaListener
    public void on(CategoryCreatedEvent event) throws Exception{
        CompletableFuture<Category> categoryCompletableFuture = catalogServiceClient.getCategory(event.getId());
        Category category = categoryCompletableFuture.get();
        IndexDocument doc = new IndexDocument(category.getId());
        doc.put(CategoryMapping.FIELD_ID, category.getId());
        doc.put(CategoryMapping.FIELD_NAME, category.getName());
        doc.put(CategoryMapping.FIELD_PARENT_ID, category.getParentId());
        doc.put(CategoryMapping.FIELD_LEVEL , category.getLevel());

        CreateDocumentCriteria createDocumentCriteria = new CreateDocumentCriteria();
        createDocumentCriteria.setIndex(CategoryMapping.INDEX);
        createDocumentCriteria.setType(CategoryMapping.TYPE);
        createDocumentCriteria.setDocument(doc);
        elasticTemplate.insert(createDocumentCriteria);
    }

}
